/*
    Copyright 2010 Northbranchlogic, Inc.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.norbl.runjrun;

import com.xerox.amazonws.sqs2.*;
import com.norbl.util.s3.*;
import com.norbl.util.sdb.*;
import com.norbl.util.sqs.*;
import com.norbl.util.ec2.*;
import com.norbl.util.*;
import java.util.*;
import java.io.*;

/**
 *
 * @author Barnet Wagman
 */
abstract public class RunnerNode {

    protected NodeStateTable nodeStateTable;
    protected NodeState nodeState;

    protected TaskStateTable taskStateTable;

    protected String host;
    protected String instanceID;

    protected NodeParams params;

    protected SqsOps sqsOps;
    protected S3Ops s3Ops;
    protected SdbOps sdbOps;
    protected Ec2Ops eops;

    protected HashMap<String,Long> missingTaskHt;

    public RunnerNode() {
        try {
            System.setProperty("org.apache.commons.logging.Log",
                               "com.norbl.util.s3.NilLog");

            params = new NodeParams(getUserData());

            sdbOps = new SdbOps(params.awsAccessID, params.awsSecretKey,
                                params.sdbRetryMillis.longValue(),
                                params.sdbMaxWaitMillis.longValue());

            s3Ops = new S3Ops(params.awsAccessID, params.awsSecretKey,
                              params.s3RetryMillis.longValue(),
                              params.s3MaxWaitMillis.longValue());

            sqsOps = new SqsOps(params.awsAccessID,
                                        params.awsSecretKey,
                                        params.taskQueueName);

            eops = new Ec2Ops(params.awsAccessID, params.awsSecretKey);

            nodeStateTable = new NodeStateTable(sdbOps,eops,
                                                params.pulseIntervalMillis);

            taskStateTable = new TaskStateTable(sdbOps);

            host = NetUtil.getHost();
            instanceID = getInstanceID();

            missingTaskHt = new HashMap<String, Long>();

            setInitialState();
        }
        catch(Throwable xxx) {
            handleThrowable(xxx);
            System.exit(0); // Exceptions in the constructor are terminal.
        }
    }

    abstract protected String getUserData();
    abstract protected void handleThrowable(Throwable xxx);
    abstract String getInstanceID();

        // ----------------------------------------------

    public void run() {

        startPulseThread();

        long lastCleanTime = 0L;

        NEXT: for (;;) {
            try {
                long tm = System.currentTimeMillis();
                if ( (tm - lastCleanTime) > params.cleanIntervalMillis.longValue() ) {
                    clean();
                    lastCleanTime = tm;
                }

                Message m = sqsOps.queue.receiveMessage();

                if ( m == null ) { // Take a little nap.
                    setStatusIdle();
                    try { Thread.sleep(params.emptyQueueNapMillis.longValue()); }
                    catch(InterruptedException ix) {}
                    continue NEXT;
                }

                TaskDescription des = new TaskDescription(m.getMessageBody());

                Runnable r = getRunnable(des);
                if ( r == null ) {                  
                    handleMissingTask(m,des);
                    setStatusIdle();
                    continue NEXT;
                }
                else {
                    deleteMessage(m);
                    setStatusActive(des);
                    tm = System.currentTimeMillis();
                    taskStateTable.setTaskStateRunning(host, instanceID, des, tm);
                }

                if ( runTask(r,des) ) {
                    tm = System.currentTimeMillis();
                    taskStateTable.setTaskStateFinished(host, instanceID, des, tm);
                    setStatusIdle();
                }
                else {
                    tm = System.currentTimeMillis();
                    taskStateTable.setTaskStateTerminatedByException(
                                                host, instanceID, des, tm);
                    setStatusIdle();
                }
            }
            catch(Throwable x) {
                if ( !((x instanceof SQSException) &&
                        SqsUtil.isNonexistentQueueException((SQSException) x)) )
                    handleThrowable(x);
                setStatusIdle();
                try { Thread.sleep(params.emptyQueueNapMillis.longValue()); }
                catch(InterruptedException ix) {}
            }
        }
    }

        // -------- State update methods -----------------

        // Note that these are in this class rather than in
        // NodeStateTable because they should only be called
        // from the node.

    void setInitialState() {

        try {
            Long tm = new Long(System.currentTimeMillis());

            nodeState = new NodeState(host,
                                      instanceID,
                                      NodeState.InstanceState.running,
                                      tm,
                                      NodeState.Status.idle,
                                      tm,
                                      null // running task
                                      );
             nodeStateTable.setState(host, nodeState, true);          
        }
        catch(Throwable x) { handleThrowable(x); }
    }

    void setStatusActive(TaskDescription des) {
        try {
            nodeState.status = NodeState.Status.active;
            nodeState.statusTime = new Long(System.currentTimeMillis());
            nodeState.runningTask = des.inputS3Key;
            nodeStateTable.setState(host, nodeState, true);           
        }
        catch(Throwable x) { handleThrowable(x); }
    }

    void setStatusIdle() {
        try {
            nodeState.status = NodeState.Status.idle;
            nodeState.statusTime = new Long(System.currentTimeMillis());
            nodeState.runningTask = null;
            nodeStateTable.setState(host, nodeState, true);
        }
        catch(Throwable x) { handleThrowable(x); }
    }

    void updateInstanceState() {
        try {
            nodeState.statusTime = new Long(System.currentTimeMillis());
            nodeStateTable.setState(host, nodeState, true);
        }
        catch(Throwable x) { handleThrowable(x); }
    }        
        // ------------------------------------------------

        /** Runs task, writes cooked to s3 and deletes raw from s3.
         *  The delete is always performed, even if something else
         *  fails.
         *
         * @param r
         * @param des
         * @return true if run and save cooked task execute without
         *  exception.
         */
    public boolean runTask(Runnable r, TaskDescription des) {
        try {
            r.run();
            saveCookedTask(r,des);
            return(true);
        }
        catch(Throwable xxx) {
            handleThrowable(xxx);
            return(false);
        }
        finally {
            deleteRawTask(des);
        }
    }

    Runnable getRunnable(TaskDescription des) {

        try {
            return( (Runnable) s3Ops.getObject(des.inputBucketName,
                                               des.inputS3Key,
                                               true) );
        }
        catch(Throwable x) {
            handleThrowable(x);
            return(null);
        }
    }

    void handleMissingTask(Message m, TaskDescription des)     {

        Long t0 = missingTaskHt.get(des.inputS3Key);
        if ( t0 == null ) {
            t0 = new Long(System.currentTimeMillis());
            missingTaskHt.put(des.inputS3Key,t0);
        }

        long dt = System.currentTimeMillis() - t0;
        if ( dt < params.maxMissingTaskMillis.longValue() ) {
                // Leave the task message in the queue
            return;
//            setStatusIdle();
        }
        else { // The task is too old - remove it from the queue
            deleteMessage(m);
//            setStatusIdle();
            try {
                taskStateTable.setTaskStateMissingRunnable(host, instanceID, des,
                                            System.currentTimeMillis());
            }
            catch(Exception sx) { handleThrowable(sx); }
        }
    }

    void deleteMessage(Message m) {

        try {
            sqsOps.queue.deleteMessage(m);
            return;
        }
        catch(Throwable x) { handleThrowable(x); }
    }

    void saveCookedTask(Runnable r, TaskDescription des) {

        if ( !des.saveCookedRunnable ) return;

        try {
            s3Ops.addObject(des.outputBucketName,des.outputS3Key,
                            (Serializable) r,
                            true);
        }
        catch(Throwable xxx) { handleThrowable(xxx); }
    }

    void deleteRawTask(TaskDescription des) {

        if ( !des.deleteRawAfterRunning ) return;

        try {
            s3Ops.deleteObject(des.inputBucketName,des.inputS3Key,
                               true);
            return;
        }
        catch(Exception xx) { handleThrowable(xx); }
    }

    void handleExceptionFromTask(TaskDescription des, Throwable x) {
        try {
            taskStateTable.setTaskStateTerminatedByException(host, instanceID, des,
                                        System.currentTimeMillis());
        }
        catch(Exception sx) { handleThrowable(sx); }
    }

    void clean() {
        sqsOps = sqsOps.cleanClone();
        s3Ops = s3Ops.cleanClone();
        sdbOps = sdbOps.cleanClone();
        eops = eops.cleanClone();

        nodeStateTable = nodeStateTable.cleanClone(sdbOps, eops);
        taskStateTable = taskStateTable.cleanClone(sdbOps);
    }

        // -------------------------------------------

    void startPulseThread() {
        (new Thread(new Pulser())).start();
    }

    class Pulser implements Runnable {

        public void run() {
            Thread.currentThread().setPriority(Thread.MAX_PRIORITY);
            for (;;) {
                updateInstanceState();
                Time.nap(params.pulseIntervalMillis.longValue());
            }
        }
    }
}
